/*
 * Copyright (C) 2005-present, 58.com.  All rights reserved.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.swak.mmap;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

import com.swak.nativex.nativeio.NativeIO;
import com.swak.utils.FileUtils;
import com.swak.utils.Lists;
import com.swak.utils.StringUtils;

import lombok.Getter;

/**
 * non-threadsafe.
 * 
 * MappedFile.
 */
public class MappedFile extends ReferenceResource {

	public static final Logger log = LoggerFactory.getLogger(MappedFile.class);
	// 操作系统的 pagesize(4k)
	private static final int OS_PAGE_SIZE = NativeIO.getOperatingSystemPageSize();
	// 默认刷盘的时机： 4kb
	private static final int Flush_Page_DNF = 1;
	// 当前JVM中映射的虚拟内存总大小
	private static final AtomicLong TotalMapedVitualMemory = new AtomicLong(0);
	// 当前JVM中mmap句柄数量
	private static final AtomicInteger TotalMapedFiles = new AtomicInteger(0);

	public static int getTotalmapedfiles() {
		return TotalMapedFiles.get();
	}

	public static long getTotalMapedVitualMemory() {
		return TotalMapedVitualMemory.get();
	}

	// 映射的文件名
	@Getter
	private final String fileName;
	// 映射的起始偏移量， 通过 filename 转换而来，例如：fileame - 000001， fileOffset - 1
	@Getter
	private final int fileOffset;
	// 映射的文件大小，定长
	@Getter
	private final int fileSize;
	// 记录的最大长度
	@Getter
	private final int maxRecordSize;
	// 映射的文件
	private final File file;
	// 当前写到什么位置
	private final AtomicInteger wrotePostion = new AtomicInteger(0);
	// Flush到什么位置
	private final AtomicInteger flushedPosition = new AtomicInteger(0);
	// 记录的映射
	private final ByteBuffer recordBuffer;
	// 最少刷新的页数
	private int flushLeastPages = Flush_Page_DNF;

	// mmap file or direct buffer and file
	private RandomAccessFile randomAccessFile;
	private FileChannel fileChannel;
	private MappedByteBuffer mappedByteBuffer;

	// 最后一条消息存储时间
	@Getter
	private volatile long lastStoreTime = 0;
	private AtomicBoolean isMapped = new AtomicBoolean(false);
	private ReentrantLock mapLock = new ReentrantLock();
	private AtomicLong lastTouchTime = new AtomicLong();

	public MappedFile(final String fileName, final int fileOffset, final int fileSize, final int maxRecordSize)
			throws IOException {
		this(fileName, fileOffset, fileSize, maxRecordSize, maxRecordSize);
	}

	public MappedFile(final String fileName, final int fileOffset, final int fileSize, final int maxRecordSize,
			int flushLeastPages) throws IOException {
		Assert.isTrue(!StringUtils.isBlank(fileName), "filename is null");
		Assert.isTrue(fileSize > 0, "please set fileSize");
		Assert.isTrue(maxRecordSize > 0, "please set maxRecordSize");
		this.fileName = fileName;
		this.fileSize = fileSize;
		this.maxRecordSize = maxRecordSize + MappedRecord.HEADER_LENGTH;
		this.file = new File(fileName);
		this.fileOffset = fileOffset;
		this.recordBuffer = ByteBuffer.allocate(this.maxRecordSize);
		this.flushLeastPages = flushLeastPages;
	}

	public MappedFile mapFile() throws IOException {
		boolean ok = false;
		FileUtils.ensureDirOK(this.file.getParent());

		if (!this.isMapped.get()) {
			try {
				mapLock.lock();
				if (!this.isMapped.get()) {
					this.randomAccessFile = new RandomAccessFile(this.file, "rw");
					this.fileChannel = this.randomAccessFile.getChannel();
					this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);

					// 默认是不能写入数据的，必须调用 手动 recover() 之后才能写入数据：
					// 1. 保护已经写入的数据不会被覆盖；
					// 2. recover 也只需要在系统启动的时候才需要调用；
					this.wrotePostion.set(this.fileSize);
					this.flushedPosition.set(this.fileSize);

					TotalMapedVitualMemory.addAndGet(fileSize);
					TotalMapedFiles.incrementAndGet();
					this.lastTouchTime.set(System.currentTimeMillis());
					ok = true;
					this.isMapped.set(true);
					log.info("map file {}.", this.fileName);
				}
			} catch (FileNotFoundException e) {
				log.error("create file channel " + this.fileName + " Failed. ", e);
				throw e;
			} catch (IOException e) {
				log.error("map file " + this.fileName + " Failed. ", e);
				throw e;
			} finally {
				mapLock.unlock();
				if (!ok && this.fileChannel != null) {
					this.fileChannel.close();
					this.randomAccessFile.close();
				}
			}
		}

		return this;
	}

	public MappedFile unMapFile() throws IOException {
		if (this.isMapped.get()) {
			try {
				mapLock.lock();
				if (this.isMapped.get()) {
					this.isMapped.set(false);
					this.mappedByteBuffer.force();
					this.flushedPosition.set(this.wrotePostion.get());

					NativeIO.freeDirectBuffer(this.mappedByteBuffer);
					TotalMapedVitualMemory.addAndGet(this.fileSize * (-1));
					TotalMapedFiles.decrementAndGet();
					log.info("unmap file REF: {} name: {} OK", this.refCount.get(), this.fileName);
					this.fileChannel.close();
					this.randomAccessFile.close();
				}
			} catch (IOException e) {
				log.error("unmap file " + this.fileName + " Failed. ", e);
				throw e;
			} finally {
				mapLock.unlock();
			}
		}
		return this;
	}

	/**
	 * 启动系统时，应该从文件系统中恢复此映射, 不然会导致数据的覆盖.
	 * 
	 * @return
	 * @throws IOException
	 */
	public MappedFile recover() throws IOException {

		// init mmap
		if (!this.isMapped.get()) {
			mapFile();
		}

		// recover
		ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
		int mapedFileOffset = 0;
		while (true) {
			int initialPos = byteBuffer.position();
			MappedRecord record = null;
			try {
				record = MappedRecord.decodeHeader(byteBuffer);
			} catch (Exception e) {
				e.printStackTrace();
			}
			if (record == null) {
				break;
			}
			byteBuffer.position(initialPos + record.recordLength());
			mapedFileOffset += record.recordLength();
		}

		int freeSpace = this.fileSize - byteBuffer.position();
		if (freeSpace < mapedFileOffset + MappedRecord.HEADER_LENGTH) {
			mapedFileOffset = this.fileSize;
		}

		this.wrotePostion.set(mapedFileOffset);
		this.flushedPosition.set(mapedFileOffset);
		return this;
	}

	/**
	 * 
	 * 需要在刚开始的收预加载, 不能轻易的调用，后面的再优化.
	 * 
	 * @param type
	 * @param pages
	 * @throws IOException
	 */
	public MappedFile warmMappedFile(FlushDiskType type, int pages) throws IOException {
		long beginTime = System.currentTimeMillis();
		ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
		int flush = 0;
		long time = System.currentTimeMillis();
		for (int i = 0, j = 0; i < this.fileSize; i += MappedFile.OS_PAGE_SIZE, j++) {

			// 每 4k 的地方写入 （byte）0，相当于预加载
			byteBuffer.put(i, (byte) 0);

			// force flush when flush disk type is sync
			if (type == FlushDiskType.SYNC_FLUSH) {
				if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {
					flush = i;
					mappedByteBuffer.force();
				}
			}

			// prevent gc
			if (j % 1000 == 0) {
				log.info("j={}, costTime={}", j, System.currentTimeMillis() - time);
				time = System.currentTimeMillis();
				try {
					// 让当前线程让出执行权，因为等待时间为0 则可以认为是进入可执行队列
					// 和 Thread.yield() 有什么区别； 在 hostspot 中是一样的
					Thread.sleep(0);
				} catch (InterruptedException e) {
					log.error("Interrupted", e);
				}
			}
		}

		// force flush when prepare init finished
		if (type == FlushDiskType.SYNC_FLUSH) {
			log.info("mapped file warm-up done, force to disk, mappedFile={}, costTime={}", this.getFileName(),
					System.currentTimeMillis() - beginTime);
			mappedByteBuffer.force();
		}
		log.info("mapped file warm-up done. mappedFile={}, costTime={}", this.getFileName(),
				System.currentTimeMillis() - beginTime);

		this.mlock();
		return this;
	}

	private void mlock() throws IOException {
		final long beginTime = System.currentTimeMillis();
		final long address = NativeIO.directBufferAddress(this.mappedByteBuffer);
		{
			int ret = NativeIO.mlock(mappedByteBuffer, this.fileSize);
			log.info("mlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret,
					System.currentTimeMillis() - beginTime);
		}

		{
			int ret = NativeIO.madvise(mappedByteBuffer, this.fileSize, NativeIO.MADV_WILLNEED);
			log.info("madvise {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret,
					System.currentTimeMillis() - beginTime);
		}
	}

	public void munlock() throws IOException {
		final long beginTime = System.currentTimeMillis();
		final long address = NativeIO.directBufferAddress(this.mappedByteBuffer);
		int ret = NativeIO.munlock(this.mappedByteBuffer, this.fileSize);
		log.info("munlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret,
				System.currentTimeMillis() - beginTime);
	}

	/**
	 * 向 MapedBuffer 追加消息
	 * 
	 * @param data
	 * @return
	 * @throws IOException
	 */
	public AppendResult appendData(final byte[] data) throws IOException {

		// touch time
		this.lastTouchTime.set(System.currentTimeMillis());

		// init mmap
		if (!this.isMapped.get()) {
			mapFile();
		}

		// 写入的位置 - 剩余的空间
		int currentPos = this.wrotePostion.get();
		int maxBlank = this.fileSize - currentPos;

		// 没有空间了
		if (currentPos >= this.fileSize) {
			return new AppendResult(AppendStatus.END_OF_FILE);
		}

		// 准备buffer
		ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
		byteBuffer.position(currentPos);

		// 写入数据的位置
		long wroteOffset = this.fileOffset * this.fileSize + currentPos;

		// 准备数据
		MappedRecord record = new MappedRecord(data);

		// 数据头部是否可以写入
		if (record.recordLength() > this.maxRecordSize) {
			return new AppendResult(AppendStatus.MESSAGE_SIZE_EXCEEDED);
		}

		// 数据头部是否可以写入
		if (record.headerLength() > maxBlank) {
			return new AppendResult(AppendStatus.END_OF_FILE);
		}

		// 写入数据结果
		AppendStatus status = AppendStatus.PUT_OK;

		// 剩余的空间空间是否可以写入数据， 如果不行能直接写入空白的数据
		if (record.recordLength() > maxBlank) {
			record = record.toBlankRecord(maxBlank);
			status = AppendStatus.END_OF_FILE;
		}

		// 写入 buffer 中
		record.encode(this.recordBuffer);

		// 写入数据
		byteBuffer.put(this.recordBuffer.array(), 0, record.recordLength());
		this.wrotePostion.addAndGet(record.recordLength());
		this.lastStoreTime = record.getTimeStamp();

		// 默认超过 4kb 则需要写盘
		this.flush(this.flushLeastPages);

		// 返回写入结果
		return new AppendResult(status, wroteOffset, record.recordLength(), record.getTimeStamp());
	}

	/**
	 * 消息刷盘
	 * 
	 * @return The current flushed position
	 */
	public int flush(final int flushLeastPages) {
		if (this.isAbleToFlush(flushLeastPages)) {
			if (this.hold()) {
				int value = getReadPosition();

				try {
					this.mappedByteBuffer.force();
				} catch (Throwable e) {
					log.error("Error occurred when force data to disk.", e);
				}

				this.flushedPosition.set(value);
				this.release();
			} else {
				log.warn("in flush, hold failed, flush offset = {}.", this.flushedPosition.get());
				this.flushedPosition.set(getReadPosition());
			}
		}
		return this.getFlushedPosition();
	}

	private boolean isAbleToFlush(final int flushLeastPages) {
		if (!this.isMapped.get()) {
			return false;
		}
		int flush = this.flushedPosition.get();
		int write = this.wrotePostion.get();

		// 如果当前文件已经写满，应该立刻刷盘
		if (this.isFull()) {
			return true;
		}

		// 只有未刷盘数据满足指定page数目才刷盘
		if (flushLeastPages > 0) {
			return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= flushLeastPages;
		}
		return write > flush;
	}

	@Override
	protected boolean cleanup(final long currentRef) {
		// 如果没有被shutdown，则不可以unmap文件，否则会crash
		if (this.isAvailable()) {
			log.error("this file[REF:{} ] name: {} have not shutdown, stop unmaping.", currentRef, this.fileName);
			return false;
		}

		// 如果已经cleanup，再次操作会引起crash
		if (this.isCleanupOver()) {
			log.error("this file[REF: {} ] name: {} have cleanup, do not do it again.", currentRef, this.fileName);
			// 必须返回true
			return true;
		}

		NativeIO.freeDirectBuffer(this.recordBuffer);
		if (!this.isMapped.get()) {
			return true;
		}
		NativeIO.freeDirectBuffer(this.mappedByteBuffer);
		TotalMapedVitualMemory.addAndGet(this.fileSize * (-1));
		TotalMapedFiles.decrementAndGet();
		log.info("unmap file[REF:{} ] name: {} OK", currentRef, this.fileName);
		return true;
	}

	/**
	 * 扫描文件.
	 * 
	 * @param address 全局的地址
	 * @param count   需要扫描的数据数量
	 * @return 扫描结果
	 * @throws IOException
	 */
	public MappedFileScan scan(int position, int count) throws IOException {

		// touch time
		this.lastTouchTime.set(System.currentTimeMillis());

		// init mmap
		if (!this.isMapped.get()) {
			mapFile();
		}

		// buffer 副本
		ByteBuffer buffer = this.mappedByteBuffer.slice();

		// scan records
		List<MappedRecord> records = Lists.newArrayList();
		buffer.position(position);
		int recordSize = 0;
		while (buffer.remaining() > 0 && recordSize < count) {
			MappedRecord record = MappedRecord.decode(buffer);

			// 没有解析到真实的记录, 空记录不算真实额记录
			if (record == null || record.isBlankRecord()) {
				break;
			}
			records.add(record);
			recordSize++;
		}
		return new MappedFileScan().setPosition(buffer.position()).setRecords(records);
	}

	public boolean isFull() {
		int freeSpace = getFreeSpace();
		return freeSpace < MappedRecord.HEADER_LENGTH;
	}

	public int getReadPosition() {
		return this.wrotePostion.get();
	}

	public int getFreeSpace() {
		return this.fileSize - this.wrotePostion.get();
	}

	public int getWrotePostion() {
		return wrotePostion.get();
	}

	public int getFlushedPosition() {
		return flushedPosition.get();
	}

	public boolean getIsMapped() {
		return isMapped.get();
	}

	public long getLastModifiedTimestamp() {
		return this.file.lastModified();
	}

	@Override
	public void shutdown(long intervalForcibly) throws IOException {

		if (this.isMapped.get()) {
			this.unMapFile();
		}

		// 释放所有的资源
		super.shutdown(intervalForcibly);

		// 释放成功
		if (this.isCleanupOver()) {
			try {
				if (this.fileChannel != null && this.fileChannel.isOpen()) {
					this.fileChannel.close();
					log.info("close file channel {} OK", this.fileName);
				}
			} catch (Exception e) {
				log.warn("close file channel " + this.fileName + " Failed. ", e);
			}
		} else {
			log.warn("destroy maped file[REF:" + this.getRefCount() + "] " + this.fileName + " Failed. cleanupOver: "
					+ this.cleanupOver);
		}
	}

	/**
	 * 清理资源，destroy与调用shutdown的线程必须是同一个
	 * 
	 * @return 是否被destory成功，上层调用需要对失败情况处理，失败后尝试重试
	 */
	public boolean destroy(final long intervalForcibly) throws IOException {

		// 释放所有的资源
		super.shutdown(intervalForcibly);

		// 释放成功
		if (this.isCleanupOver()) {
			try {
				if (this.fileChannel != null && this.fileChannel.isOpen()) {
					this.fileChannel.close();
					log.info("close file channel {} OK", this.fileName);
				}

				long beginTime = System.currentTimeMillis();
				boolean result = this.file.delete();
				log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
						+ (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePostion() + " M:"
						+ this.getFlushedPosition() + ", " + (System.currentTimeMillis() - beginTime));

				return result;
			} catch (Exception e) {
				log.warn("close file channel " + this.fileName + " Failed. ", e);
			}
			return true;
		} else {
			log.warn("destroy maped file[REF:" + this.getRefCount() + "] " + this.fileName + " Failed. cleanupOver: "
					+ this.cleanupOver);
		}
		return false;
	}

	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append("fileName=").append(this.fileName).append(";")// fileName
				.append("wrotePostion=").append(getWrotePostion()).append(";") // wrotePostion
				.append("flushedPosition=").append(getFlushedPosition()).append(";") // flushedPosition
				.append("fileSize=").append(this.getFileSize()).append(";")// fileSize
				.append("freeSpace=").append(this.getFreeSpace()).append(";");// freeSpace
		return builder.toString();
	}
}
